package com.sduept.nwld.dataserver.manager.fault;

import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.sduept.bigdata.fault.common.ParamDesc;
import com.sduept.bigdata.fault.entity.FaultRecord;
import com.sduept.bigdata.fault.entity.RequestTaskRecord;
import com.sduept.bigdata.fault.manager.FaultManager;
import com.sduept.bigdata.fault.manager.RequestTaskManager;
import com.sduept.core.ParamManager;
import com.sduept.core.dao.SuperDAO;
import com.sduept.core.entity.SystemParameter;
import com.sduept.kafka.service.KafkaMessageSender;
import com.sduept.kafka.service.exception.NoSuchTopicException;
import com.sduept.nwld.dataserver.manager.etl.ProtectionDeviceActionETLManager;
import com.sduept.utils.DateUtil;

import net.sf.json.JSONObject;

/**
 * 该类用来获取数据 ，同时将任务保存在表中
 */
@Service
public class FaultTimingTaskService {

    @Autowired
    private SuperDAO dao;
    @Autowired
    private ProtectionDeviceActionETLManager actionETLManager;
    @Autowired
    private FaultManager faultManager;
    @Autowired
    private RequestTaskManager timingManager;
    @Autowired
    private ParamManager cache;
    @Autowired
    private KafkaMessageSender kafkaMessageSender;
    private static ExecutorService executor = Executors.newFixedThreadPool(20);

    private boolean run = false;

    private ConcurrentMap<String, String> faultIdsMap = new ConcurrentHashMap<>();// 用来去除重复的故障ID

    private ConcurrentMap<String, String> faultTimesMap = new ConcurrentHashMap<>();// 用来去除重复的故障时间

    private Object faultIdObjectLock = new Object();

    private Object faultTimeObjectLock = new Object();

    // todo springboot定时任务
    @Scheduled(cron = "1 1 0/4 * * ?")
    public void doFaultTask() {
        List<RequestTaskRecord> list = timingManager.listRecordsByStatus(0);
        doFaultTimingTask(list, false);
    }

    private void doSingleFaultTimingTask(final RequestTaskRecord record, final boolean neglectingConditions, final int dayTerm) {
        // 动作元件
        if (0 == record.getType()) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    record.setTaskStatus(1);
                    FaultRecord fault = dao.findById(record.getTaskId(), FaultRecord.class);
                    if (fault == null) {
                        return;
                    }
                    // 如果不是故障
                    if (fault.getStatus() == 1) {
                        return;
                    }
                    if (null == fault.getFaultTime()) {
                        return;
                    }
                    if (null == record.getStartTime()) {
                        return;
                    }
                    DateTime recordTime = new DateTime(record.getStartTime());
                    DateTime temp = recordTime.plusDays(dayTerm);
                    DateTime temp1 = recordTime.plusHours(2);
                    // 如果超过期限，直接返回
                    if (!record.isNeglectingConditions() || !neglectingConditions) {
                        if (temp.isBeforeNow()) {
                            record.setTaskStatus(3);
                            record.setEndTime(new Date());
                            timingManager.saveRecord(record);
                            return;
                        }
                    }
                    if (temp1.isBeforeNow() || neglectingConditions) {
                        String datetime = DateUtil.getStringDate(fault.getFaultTime(), DateUtil.pattern_ymdhms);
                        synchronized (faultTimeObjectLock) {
                            String sign = faultTimesMap.get(datetime);
                            if (StringUtils.isEmpty(sign)) {
                                faultTimesMap.put(datetime, datetime);
                                regetProtectActions(datetime);

                            }
                        }
                    }
                    timingManager.saveRecord(record);
                    actionETLManager.getAndSaveActionData(record.getStationId(), fault);
                }
            });
        } else if (1 == record.getType()) {
            // 录波文件列表
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    FaultRecord faultRecord = dao.findById(record.getTaskId(), FaultRecord.class);
                    if (null == faultRecord) {
                        return;
                    }
                    // 如果不是故障
                    if (faultRecord.getStatus() == 1) {
                        return;
                    }
                    if (null == faultRecord.getFaultTime()) {
                        return;
                    }
                    if (null == record.getStartTime()) {
                        return;
                    }
                    DateTime faultTime = new DateTime(record.getStartTime());
                    DateTime temp = faultTime.plusDays(dayTerm);
                    DateTime temp1 = faultTime.plusHours(2);
                    if (StringUtils.isNotEmpty(faultRecord.getEquipment_id())) {
                        if (!record.isNeglectingConditions() || !neglectingConditions) {
                            if (temp.isBeforeNow()) {
                                record.setTaskStatus(3);
                                record.setEndTime(new Date());
                                timingManager.saveRecord(record);
                                return;
                            }
                        }
                        // 20180205 新增判断，当故障生成2小时之内没有录波文件推送，再去发消息执行获取
                        if (temp1.isBeforeNow() || neglectingConditions) {
                            record.setTaskStatus(1);
                            timingManager.saveRecord(record);
                            synchronized (faultIdObjectLock) {
                                String sign = faultIdsMap.get(record.getTaskId());
                                if (StringUtils.isEmpty(sign)) {
                                    faultIdsMap.put(record.getTaskId(), record.getTaskId());
                                    String datetime = DateUtil.getStringDate(faultRecord.getFaultTime(), DateUtil.pattern_ymdhms);
                                    try {
                                        faultManager.regetWaveAndParse(datetime, record.getTaskId());
                                    } catch (NoSuchTopicException e) {
                                        e.printStackTrace();
                                    } catch (JsonProcessingException e) {
                                        e.printStackTrace();
                                    }

                                }
                            }
                        }
                    }
                }
            });
        }

    }

    /**
     * 执行故障数据获取任务
     *
     * @param list
     * @param neglectingConditions 忽略条件限制
     */
    public void doFaultTimingTask(List<RequestTaskRecord> list, boolean neglectingConditions) {
        // 如果正在还有任务未执行完成，直接跳过当次任务
        if (run) {
            return;
        }
        run = true;
        SystemParameter parameter = cache.get(ParamDesc.FAULT_DATA_GET_TERM);
        int dayTerm = 30;
        if (parameter != null && StringUtils.isNotEmpty(parameter.getValue())) {
            dayTerm = Integer.parseInt(parameter.getValue());
        }
        faultIdsMap.clear();
        faultTimesMap.clear();
        for (final RequestTaskRecord record : list) {
            doSingleFaultTimingTask(record, neglectingConditions, dayTerm);
        }
        run = false;
    }

    public void regetProtectActions(String dateTime) {
        MsgContent msg = new MsgContent(dateTime, "");
        try {
            kafkaMessageSender.sendMessage("sf_protect_action_reget", JSONObject.fromObject(msg));
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        } catch (NoSuchTopicException e) {
            e.printStackTrace();
        }
    }

    public class MsgContent {
        String updatetime;
        String requestId;

        public MsgContent(String updatetime, String requestId) {
            super();
            this.updatetime = updatetime;
            this.requestId = requestId;
        }

        public String getUpdatetime() {
            return updatetime;
        }

        public void setUpdatetime(String updatetime) {
            this.updatetime = updatetime;
        }

        public String getRequestId() {
            return requestId;
        }

        public void setRequestId(String requestId) {
            this.requestId = requestId;
        }
    }

}
